package cn.gwm.flink.streaming.function.process;

import cn.gwm.flink.streaming.beans.termalRunaway.*;
import cn.gwm.flink.streaming.constant.TermalProcessResult;
import cn.gwm.flink.streaming.constant.TermalRunawayTypeEnum;
import cn.gwm.flink.streaming.constant.VehiclePowerType;
import cn.gwm.flink.streaming.strategy.termal.TermalStrategyFactory;
import cn.gwm.utils.StreamUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.lang.Dict;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.json.JSONUtil;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 热失控数据处理函数
 *
 * @author GW00283474
 */
public class TermalWarningPhevProcessFunction extends KeyedProcessFunction<String, WaringTermalDwmWithStragegyPhev, Dict> {
    private static final Logger logger = LoggerFactory.getLogger(TermalWarningPhevProcessFunction.class);
    /**
     * 温度校验维度的信号集合状态
     */
    private ListState<WaringTermalDwmWithStragegyPhev> temperatureState;
    /**
     * 电压校验维度的信号集合状态
     */
    private ListState<WaringTermalDwmWithStragegyPhev> voltageState;
    /**
     * 温差校验维度的信号集合状态
     */
    private ListState<WaringTermalDwmWithStragegyPhev> temperatureDiffState;
    /**
     * 注册的定时器对应的校验策略集合
     */
    private MapState<Long, List> timeStrategyState;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        StateTtlConfig stateTtlConfig = StateTtlConfig
                // 状态有效时间    1天过期
                .newBuilder(Time.days(1))
                // 设置状态的更新类型
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                // 已过期还未被清理掉的状态数据不返回给用户
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                // 过期对象的清理策略 全量清理
                .cleanupFullSnapshot()
                .build();
        ListStateDescriptor temperatureDesc = new ListStateDescriptor("temperatureDescPhev", WaringTermalDwmWithStragegyPhev.class);
        temperatureDesc.enableTimeToLive(stateTtlConfig);
        temperatureState = getRuntimeContext().getListState(temperatureDesc);
        ListStateDescriptor voltageDesc = new ListStateDescriptor("voltageDescPhev", WaringTermalDwmWithStragegyPhev.class);
        voltageDesc.enableTimeToLive(stateTtlConfig);
        voltageState = getRuntimeContext().getListState(voltageDesc);
        ListStateDescriptor temperatureDiffDesc = new ListStateDescriptor("temperatureDiffDescPhev", WaringTermalDwmWithStragegyPhev.class);
        temperatureDiffDesc.enableTimeToLive(stateTtlConfig);
        temperatureDiffState = getRuntimeContext().getListState(temperatureDiffDesc);
        MapStateDescriptor timerDesc = new MapStateDescriptor("timerDescPhev", Long.class, List.class);
        timerDesc.enableTimeToLive(stateTtlConfig);
        timeStrategyState = getRuntimeContext().getMapState(timerDesc);
    }

    @Override
    public void processElement(WaringTermalDwmWithStragegyPhev record, KeyedProcessFunction<String, WaringTermalDwmWithStragegyPhev, Dict>.Context ctx, Collector<Dict> out) throws Exception {
        try {
            List<BmsTrdStrategys> bmsTrdStrategies = record.getBmsTrdStrategies();
            if(CollectionUtil.isNotEmpty(bmsTrdStrategies)){
                //温度校验策略
                BmsTrdStrategys temperatureStrategy = bmsTrdStrategies.stream().filter(s -> "1".equals(s.getState()) && TermalRunawayTypeEnum.TEMPERATURE.getKey().equals(s.getItemType()))
                        .findFirst().orElseGet(() -> {
                            return null;
                        });
                //电压校验策略
                BmsTrdStrategys voltageStrategy = bmsTrdStrategies.stream().filter(s -> "1".equals(s.getState()) && TermalRunawayTypeEnum.VOLTAGE.getKey().equals(s.getItemType()))
                        .findFirst().orElseGet(() -> {
                            return null;
                        });
                //温差校验策略
                BmsTrdStrategys temperatureDiffStrategy = bmsTrdStrategies.stream().filter(s -> "1".equals(s.getState()) && TermalRunawayTypeEnum.TEMPERATUREDIFFERENCE.getKey().equals(s.getItemType()))
                        .findFirst().orElseGet(() -> {
                            return null;
                        });
                if (Objects.nonNull(temperatureStrategy)) {
                    temperatureState.add(record);
                    //注册温度校验定时器
                    long time = record.getItem_time() + temperatureStrategy.getDuration()*1000L;
                    ctx.timerService().registerEventTimeTimer(time);
                    mapStateHandle(timeStrategyState, time, TermalRunawayTypeEnum.TEMPERATURE.getKey());
                }
                if (Objects.nonNull(voltageStrategy)) {
                    voltageState.add(record);
                    //注册电压校验定时器
                    long time = record.getItem_time() + voltageStrategy.getDuration()*1000L;
                    ctx.timerService().registerEventTimeTimer(time);
                    mapStateHandle(timeStrategyState, time, TermalRunawayTypeEnum.VOLTAGE.getKey());
                }
                if (Objects.nonNull(temperatureDiffStrategy)) {
                    temperatureDiffState.add(record);
                    //注册温差校验定时器
                    long time = record.getItem_time() + temperatureDiffStrategy.getDuration()*1000L;
                    ctx.timerService().registerEventTimeTimer(time);
                    mapStateHandle(timeStrategyState, time, TermalRunawayTypeEnum.TEMPERATUREDIFFERENCE.getKey());
                }
                //System.out.println("vin:"+ctx.getCurrentKey()+"  定时器列表："+StreamUtil.convertIterable(timeStrategyState.keys()));
            }

        } catch (Exception e) {
            e.printStackTrace();
            logger.error("热失控检测的温差、温度、电压、车端上传的检测异常message：{},case:{},stackTrace:{}",e.getMessage(),e.getCause(),e.getStackTrace());
        }

    }

    @Override
    public void onTimer(long timestamp, KeyedProcessFunction<String, WaringTermalDwmWithStragegyPhev, Dict>.OnTimerContext ctx, Collector<Dict> out) throws Exception {
        //System.out.println("key:"+ctx.getCurrentKey()+"   定时器："+timestamp+" 触发");
        ConcurrentHashMap<String,ListState> listStateMap=new ConcurrentHashMap<>(0);
        listStateMap.put(TermalRunawayTypeEnum.TEMPERATURE.getKey(),temperatureState);
        listStateMap.put(TermalRunawayTypeEnum.VOLTAGE.getKey(),voltageState);
        listStateMap.put(TermalRunawayTypeEnum.TEMPERATUREDIFFERENCE.getKey(),temperatureDiffState);
        List<String> strategies = (List<String>) timeStrategyState.get(timestamp);
        //System.out.println("key:"+ctx.getCurrentKey()+"   该定时器对应的校验策略keys:"+strategies);
        List<Dict> dictList=new ArrayList<>(0);
        if(CollectionUtil.isNotEmpty(strategies)){
            strategies.forEach(strategy->{
                Dict dict=TermalStrategyFactory.getTermalStrategy(strategy,true, VehiclePowerType.PHEV).check(listStateMap.get(strategy),timestamp);
                if(Objects.nonNull(dict)&& (!dict.isEmpty())){
                    dictList.add(dict);
                }
            });
        }
        //System.out.println("key:"+ctx.getCurrentKey()+"  移除定时器："+timestamp);
        ctx.timerService().deleteEventTimeTimer(timestamp);
        List<Long> oldTimers = StreamUtil.convertIterable(timeStrategyState.keys()).stream().filter(s -> NumberUtil.isLessOrEqual(new BigDecimal(s), new BigDecimal(timestamp))).collect(Collectors.toList());
        //移除过期定时器
        for (Long timer: oldTimers) {
            if(!timeStrategyState.isEmpty()){
                timeStrategyState.remove(timer);
            }
        }
        WaringTermalWithStragegyAndFlagPhev record=null;
        Dict result=new Dict();
        for (Dict dict:dictList){
            if(Objects.isNull(record)){
               record=(WaringTermalWithStragegyAndFlagPhev)dict.get(TermalProcessResult.CHECK_RECORD);
            }
            if(TermalRunawayTypeEnum.TEMPERATURE.getKey().equals(dict.get(TermalProcessResult.STRATEGY_KEY))){
                record.setTemperatureWarn(Boolean.valueOf(dict.get(TermalProcessResult.MATCH).toString()));
                result.set(TermalProcessResult.DICT_TEMP,dict);
            }
            if(TermalRunawayTypeEnum.VOLTAGE.getKey().equals(dict.get(TermalProcessResult.STRATEGY_KEY))){
                record.setVoltageWarn(Boolean.valueOf(dict.get(TermalProcessResult.MATCH).toString()));
                result.set(TermalProcessResult.DICT_VOLTAGE,dict);
            }
            if(TermalRunawayTypeEnum.TEMPERATUREDIFFERENCE.getKey().equals(dict.get(TermalProcessResult.STRATEGY_KEY))){
                record.setTemperatureDifferenceWarn(Boolean.valueOf(dict.get(TermalProcessResult.MATCH).toString()));
                result.set(TermalProcessResult.DICT_TEMPDIFF,dict);
            }
        }
        if(Objects.nonNull(record) && !record.checkFieldNullValue()){
            result.set(TermalProcessResult.CHECK_RECORD,record);
            logger.info("vin:{},item_time:{}通过温度、电压、温差校验;信号为：{}",record.getDeviceId(),record.getItem_time(), JSONUtil.toJsonPrettyStr(record));
            out.collect(result);
        }
    }

    /**
     * mapState添加元素处理
     *
     * @param mapState map状态
     * @param key      map主键
     * @param element  list中的元素
     * @throws Exception
     */
    private void mapStateHandle(MapState<Long, List> mapState, Long key, String element) throws Exception {
        boolean contains = mapState.contains(key);
        if (contains) {
            List<String> list = (List<String>) mapState.get(key);
            //System.out.println("定时器："+key+",已存在策略："+list);
            if (!list.contains(element)) {
                list.add(element);
                mapState.put(key, list);
            }
            //System.out.println("定时器："+key+",已填充后的策略："+list);
        } else {
            List list = new ArrayList<String>(0);
            list.add(element);
            mapState.put(key, list);
        }
    }
}
